/*
 * ByteBufferInputQueue.java    0.0.1    18/set/2009
 * 
 * Copyright (c) 2009 mentalsmash.org
 * 
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 * The use of the Apache License does not indicate that this project is
 * affiliated with the Apache Software Foundation.
 */ 
package org.mentalsmash.tazio.net.cs.server.impl;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.ReadableByteChannel;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Implementation of a chanel's Input queue, using a ByteBuffer as back-end
 * <p>
 * This class and all the classes in this package are largely inspired to the presentation
 * "How To Build A Scalable NIO Server, Mark II" by Ron Hitchens (http://ronsoft.com)
 * </p>
 * @version    0.0.1    18/set/2009    
 * @author     Andrea Reale <andrea.reale@gmail.com>
 */
public class ByteBufferInputQueue implements InputQueue {
	/*
	 * Every method in this class is synchronised since a channel can potentially be
	 * accessed by more then one thread 
	 */
	
	private final static Logger log = LoggerFactory.getLogger(ByteBufferInputQueue.class);
	private final ByteBuffer emptyBuffer = ByteBuffer.allocate(0).asReadOnlyBuffer();
	private final BufferPool bufferPool;
	private ByteBuffer buffer;
	
	public ByteBufferInputQueue(BufferPool pool){
		this.bufferPool = pool;
	}
	
	
	/* (non-Javadoc)
	 * @see org.mentalsmash.tazio.net.server.InputQueue#dequeueBytes(int)
	 */
	@Override
	public synchronized ByteBuffer dequeueBytes(int count) {
		int howMuchToTake;
		
		if ((buffer == null) || (buffer.position() == 0) || (count == 0)) {
			return this.emptyBuffer;
		}
		
		howMuchToTake = Math.min(count, buffer.position());
		
		//Not a buffer from the pool since this is not a direct buffer, so it is quite unexpensive
		ByteBuffer toReturn = ByteBuffer.allocate(howMuchToTake);

		buffer.flip(); //buffer in "read-mode"
		if(buffer.remaining() <= toReturn.remaining()) {
			toReturn.put(buffer);
		} else {
			while (toReturn.hasRemaining()) {
				toReturn.put(buffer.get());
			}
		}
		
		if(buffer.remaining() == 0){
			//if the input buffer is empty return it to the pool
			try {
				bufferPool.returnObject(buffer);
			} catch (Exception e) {
				log.warn("Unable to return empty buffer",e);
				throw new BufferException(e);
			}
			
			buffer = null;
		} else {
			buffer.compact(); //discard dequeued bytes and set in "write-mode"
		}
		
		toReturn.flip(); //"read-mode"
		return toReturn;
	}

	/* (non-Javadoc)
	 * @see org.mentalsmash.tazio.net.server.InputQueue#fillFrom(java.nio.channels.ByteChannel)
	 */
	@Override
	//This is synchronised because multiple thread may be accessing the same channel at one time
	public synchronized int fillFrom(ReadableByteChannel source) throws IOException {
		
		if(buffer == null) {
			bufferFromPool();
		}
		
		return source.read(buffer);
	}


	/**
	 * Retrieves a buffer from the pool, makes the appropriate validations, and sets it
	 * on the instance variable
	 */
	private void bufferFromPool() {
		try {
			buffer = bufferPool.borrowObject();
		} catch (Exception e) {
			log.warn("Unable to borrow a buffer",e);
			throw new BufferException(e);
		} finally {
			if (buffer != null){
				try {
					bufferPool.returnObject(buffer);
				} catch (Exception e) {
					//just ignore
				}
			}
		}
	}

	/* (non-Javadoc)
	 * @see org.mentalsmash.tazio.net.server.InputQueue#indexOf(byte)
	 */
	@Override
	public synchronized int indexOf(byte b) {
		int pos = 0;
		
		if (buffer == null) {
			return -1;
		}

		 pos = buffer.position();

		for (int i = 0; i < pos; i++) {
			if (b == buffer.get (i)) {
				return i;
			}
		}

		return -1;
	}

	/* (non-Javadoc)
	 * @see org.mentalsmash.tazio.net.server.InputQueue#isEmpty()
	 */
	@Override
	public synchronized boolean isEmpty() {
			return (buffer == null) || (buffer.position() == 0);
	}


	/* (non-Javadoc)
	 * @see org.mentalsmash.tazio.net.cs.server.impl.InputQueue#size()
	 */
	@Override
	public synchronized int size() {
	    	if(this.isEmpty()) {
	    	    return 0;
	    	} else {
	    	    return buffer.position();
	    	}
	}


	/* (non-Javadoc)
	 * @see org.mentalsmash.tazio.net.cs.server.impl.InputQueue#peek(int)
	 */
	@Override
	public synchronized byte[] peek(int count) {
	    throw new UnsupportedOperationException("Not yet implemented");
	    
	    //this code simply doesn't work
//		byte[] copy;
//		
//		if (this.size() < count){
//			return null;
//		}
//		
//		copy = new byte[count];
//		
//		buffer.mark(); //marks the buffer's position to make it possible to reset him after reading
//		               //count bytes
//		
//		buffer.get(copy);
//		buffer.reset();
//		
//		return copy;
	}

}
